package Queue;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.*;

/**
 * @author Xu,
 * @data 2021/1/3，
 * @time 17:23
 */
public class Queue {

    public static LinkedBlockingDeque<Long> blockingQueue = new LinkedBlockingDeque<Long>();

    public Queue() { }

    public static void main(String[] args) throws IOException {
        int port = 10303;
        ServerSocket server = new ServerSocket(port);
        ExecutorService pool =  Executors.newFixedThreadPool(100);
        while(true) {
            Socket socket = server.accept();
            Runnable runnable =() -> {
                try {
                    InputStream inputStream = socket.getInputStream();
                    byte[] bytes = new byte[1024*60];
                    int len;
                    StringBuilder stringBuilder = new StringBuilder();
                    while((len = inputStream.read(bytes)) != -1 ) {
                        stringBuilder.append(new String(bytes,0,len,"UTF-8"));
                        String string = stringBuilder.toString();
                        long l = 0;
                        String[] strings = string.split(" ");
                        String msg = "";
                        if(strings[0].equals("p")) {
                            for(int i=1;i<strings.length;i++) {
                                blockingQueue.put(Long.parseLong(strings[i]));
                                //System.out.println("传递消息：" + strings[i] + ", 队列剩余消息：" + blockingQueue.size());
                            }
                        }else {
                            OutputStream out = socket.getOutputStream();
                            for(int i=0;i<1000;i++){
                                l = blockingQueue.take();
                                msg += l + " ";
                                //System.out.println("                       消费消息：" + l + ", 队列剩余消息：" + blockingQueue.size());
                            }
                            out.write(msg.getBytes("UTF-8"));
                            socket.shutdownOutput();
                        }

                    }
                    inputStream.close();
                    socket.close();
                } catch (Exception e) {
                    e.printStackTrace();
                    // TODO: handle exception
                }
            };
            pool.submit(runnable);
        }
    }

}